package PassengerVehicle

import java.io.{BufferedReader, FileReader}
import java.util.Properties

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, SparkSession}

import scala.collection.immutable.HashMap

/**
  * 计算运力配置
  *
  * 四个纬度
  *   市内
  *   跨市
  *   省内
  *   省外
  *
  *   四月份
  *     incity=486609
  *     outcity=110339
  *     outprovince=10455
  *     607403
  */
object ylpzs {



  def main(args: Array[String]): Unit = {
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    val path = "/Users/chuxiaowei/IdeaProjects/两客一货/dataprocess/src/main/resources/base.csv"
    val cdict = readFile("/Users/chuxiaowei/IdeaProjects/两客一货/dataprocess/src/main/resources/incity.txt")
    val pdict = readFile("/Users/chuxiaowei/IdeaProjects/两客一货/dataprocess/src/main/resources/inprovince.txt")

    val spark = SparkSession.builder().master("local[4]").appName("ss").config("spark.debug.maxToStringFields", "2000").getOrCreate()
    val frame = spark.read.format("org.apache.spark.sql.execution.datasources.csv.CSVFileFormat").option("header", true).option("delimiter", ",").load(path)
    frame.registerTempTable("p")

    val prop: Properties = new Properties()
    prop.setProperty("user", "root")
    prop.setProperty("password", "123456")
    val vehicleDF: DataFrame = spark.read.jdbc("jdbc:mysql://127.0.0.1:3306/2k1w?useUnicode=true&characterEncoding=UTF-8", "zhjg_zy_vehicle", prop).filter("VEH_STATUS = 10")
    val vMap = vehicleDF.rdd.map(row => (row.getAs[String]("VEH_NO"),row.getAs[java.math.BigDecimal]("SEAT_COUNT"))).collect().toMap


    val filteredframe = spark.sql("select XL,DDZ,FCRQSJ,SCKCCPHM from p where SCKCCPHM like '鲁F%' or SCKCCPHM like '鲁Y%' or SCKCCPHM is null")
    val incityValue = filteredframe.filter(row => {
      val xl = row.getString(1)
      val ddz = row.getString(2)
      matchDDZ(xl, ddz, cdict)
    }).registerTempTable("b")

    val bframe = spark.sql("select XL, SCKCCPHM, count(1) as checi from b group by XL, SCKCCPHM")
    val decimal = bframe.rdd.map(row => {
      vMap.get(row.getAs[String]("SCKCCPHM")).getOrElse(new java.math.BigDecimal(50)).multiply(new java.math.BigDecimal(row.getAs[Long]("checi")))
    }).reduce((x, y) => x.add(y))
    println(decimal)






//    val outcityValue = filteredframe.rdd.filter(row => {
//      val xl = row.getString(1)
//      val ddz = row.getString(2)
//      !matchDDZ(xl, ddz, cdict) && matchDDZ(xl, ddz, pdict)
//    }).count()
//
//
//    val outprovinceValue = filteredframe.rdd.filter(row => {
//      val xl = row.getString(1)
//      val ddz = row.getString(2)
//      !matchDDZ(xl, ddz, pdict)
//    }).count()
//
//
//    println("incity=" + incityValue)
//    println("outcity=" + outcityValue)
//    println("outprovince=" + outprovinceValue)
//
//    println(incityValue + outcityValue + outprovinceValue)


  }



  def readFile(path: String):String = {
    val reader = new FileReader(path)
    val bufferReader = new BufferedReader(reader)
    val pdict = bufferReader.readLine()
    bufferReader.close()
    reader.close()
    pdict
  }


  def matchDDZ(xl: String, ddz: String, dict: String): Boolean = {
    val reg1 = "[0-9]"
    val reg2 = "（[\\u4e00-\\u9fa5]*）"
    val reg3 = "[(][0-9\\u4e00-\\u9fa5]*[)]"
    val reg4 = "服务站|下道车|机场|国际机场|车站|汽车站|长途车站|长途汽车站|中心站|中心|高速学生|高速|补|路口|总站|南站|北站|东站|西站|汽车"

    val xllen = xl.split("-").length
    if(dict.indexOf(ddz.replaceAll(reg1,"").replaceAll(reg2,"").replaceAll(reg3,"").replaceAll(reg4,"").trim) > -1) {
      true
    }else if(xllen == 2 && dict.indexOf(xl.split("-")(1).replaceAll(reg1,"").replaceAll(reg2,"").replaceAll(reg3,"").replaceAll(reg4,"").trim) > -1){
      true
    }else if (xllen == 1 && dict.indexOf(xl.replaceAll(reg1,"").replaceAll(reg2,"").replaceAll(reg3,"").replaceAll(reg4,"").trim) > -1) {
      true
    }else if (xllen == 3 && dict.indexOf(xl.split("-")(2).replaceAll(reg1,"").replaceAll(reg2,"").replaceAll(reg3,"").replaceAll(reg4,"").trim) > -1) {
      true
    }else {
      false
    }
  }

}
